/********************************************************/
/*	Copyright (C) 2016 Gong Li Bin			*/
/*	Project:	MapReduce			*/
/*	Author:		gong_libin			*/
/*	Date:		2016_06_01			*/
/*	File:		MrProcess.h			*/
/********************************************************/

#ifndef	_MRPROCESS_H
#define	_MRPROCESS_H 

#include "MrJob.h"
#include "MrUtils.h"

#define	MR_PR_CRTE			0x01
#define	MR_PR_SEND			0x02
#define	MR_PR_RESP			0x04
#define	MR_PR_RECV			0x08
#define	MR_PR_FAIL			0x10

#define	MR_PR_VERSION			(char*)"MrProcess-2.0.0"

#define	MR_PR_PORT			(char*)"SELECT Port FROM %s.%s WHERE HostID=%s"
#define	MR_PR_POLL			(char*)"SELECT HostID,Addr,Port FROM %s.%s where Type=0"

#define	MR_PR_HUB			(char*)"SELECT HostID,Addr,Hub FROM %s.%s where Type=4"
#define	MR_PR_MAP			(char*)"SELECT HostID,Addr,Port FROM %s.%s where Type=3"
#define	MR_PR_RED			(char*)"SELECT HostID,Addr,Port FROM %s.%s where Type=4"

typedef struct tagMrPrPacket
{
	MRJBADDR_U m_unAddr;
	MRJBPACKET_S m_stPacket;
}MRPRPACKET_S;

#define	MR_PR_HOST			0x04
typedef struct tagMrPrTask
{
	UCHAR m_ucID;
	UCHAR m_ucCur;
	CMrUdp m_CUdp;
	UCHAR m_ucStatus;
	MRJBPACKET_S m_stLocal;
	UINT m_uiHost[MR_PR_HOST];
}MRPRTASK_S;

#define	MR_PR_TASK			0x10
typedef struct tagMrPrJob
{
	char* m_pszAdp;
	CMrSet m_CHub;
	UCHAR m_ucTotal;
	MRJBPACKET_S m_stGlobal;
	MRPRTASK_S m_stTask[MR_PR_TASK];
}MRPRJOB_S;

#define	MR_PR_JOBLEN			m_stPacket.m_iLength
#define	MR_PR_JOBPKT			m_stPacket.m_uszPacket

#define	MR_PR_LCLLEN			m_stLocal.m_iLength
#define	MR_PR_LCLPKT			m_stLocal.m_uszPacket

#define	MR_PR_MRLEN			m_stGlobal.m_iLength
#define	MR_PR_MRPKT			m_stGlobal.m_uszPacket

#define	MR_PR_CURMAP			pstMap->m_stTask[ucMap]
#define	MR_PR_CURRED			pstRed->m_stTask[ucRed]
#define	MR_PR_CURJOB			pstJob->m_stTask[ucJob]

#define	MR_PR_CURMLL			MR_PR_CURMAP.MR_PR_LCLLEN
#define	MR_PR_CURMLP			MR_PR_CURMAP.MR_PR_LCLPKT
#define	MR_PR_CURRLL			MR_PR_CURRED.MR_PR_LCLLEN
#define	MR_PR_CURRLP			MR_PR_CURRED.MR_PR_LCLPKT

#define	MR_PR_TOTALPP			CPro.MrProPutTotal(CPro.MrProGetTotal() + 1)

void MrProcessStatics();
void MrProcessDynamic(void* pPara);
void* MrProcessThread(void* pPara);
void MrProcessRedSort(MRPRJOB_S* pstRed);
void MrProcessError(char* pszAdp, CMrPro& CPro);
int MrProcessConfig(char* pszConf, char* pszPort);
void MrProcessHubSort(CMrUdp& CUdp, CMrSet& CHub);
void MrProcessThreadCore(MRPRPACKET_S* pstPacket);
void MrProcessRedInit(MRPRJOB_S* pstJob, UINT* puiRed);
int MrProcessHubInit(CMrDb& rCDb, char* pszSql, MRCONFIG_S* pstConf, CMrSet* pCSet);

int MrProcessJobRecv(MRPRJOB_S* pstJob);
int MrProcessJobCheck(MRPRJOB_S* pstJob);
int MrProcessJobSend(MRPRJOB_S* pstJob, CMrSet& CDev);
bool MrProcessJobStatus(MRPRJOB_S* pstJob, UCHAR ucStatus);

int MrProcessMapCreate(CMrPro& CPro, MRPRJOB_S* pstMap);
int MrProcessRedCreate(MRPRJOB_S* pstMap, MRPRJOB_S* pstRed);
int MrProcessResTrack(UCHAR* puszResult, int iResult, UINT* puiRed);
int MrProcessResCreate(MRPRJOB_S* pstJob, UCHAR* puszResult, int* piResult);

#endif /* _MRPROCESS_H */
